Flink: Introduce Flink InputFormat #1346

JingsongLi · 2020-08-17T06:22:54Z

This is subtask of #1293

Introduce FlinkInputFormat to read records from Iceberg table by Flink engine. The FlinkInputFormat is the foundation of SQL reader and streaming reader.

Implemented by:

Introduce RowDataIterator and DataIterator, just like Spark BaseDataReader and RowDataReader.
The DataIterator provides a union iterator to union FileScanTask iterators.
Introduce FlinkSplitGenerator, this generator just wrap Iceberg CombinedScanTask to FlinkInputSplit.
Introduce FlinkInputFormat, it provides a simple API: FlinkInputFormat.builder().tableLoader(..).build().

This PR is based on:

Introduce CatalogLoader. ~~Flink: Introduce CatalogLoader and TableLoader #1332~~
Introduce GenericAppenderFactory and GenericAppenderHelper for reusing and testing. ~~Introduce GenericAppenderFactory and GenericAppenderHelper #1340~~

openinx · 2020-08-18T07:06:49Z

flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

+    }
+  }
+
+  abstract CloseableIterator<T> nextTaskIterator(FileScanTask scanTask) throws IOException;


How about renaming this method to openTaskIterator ?

openinx · 2020-08-18T07:29:09Z

flink/src/main/java/org/apache/iceberg/flink/source/RowDataIterator.java

+
+  @Override
+  protected CloseableIterator<RowData> nextTaskIterator(FileScanTask task) {
+    // schema or rows returned by readers


nit: is this comment still valuable ? Seems I did not get the point.

We can delete it.

openinx · 2020-08-18T07:52:24Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java

+  }
+
+  @Test
+  public void testProjection() throws Exception {


I'm not quite sure whether flink support complex data type projection, if sure we may need more unit tests to address the projection cases, such as projection by a nested struct, map, list (similar to the spark's TestReadProjection ).

Another case: Project with a new renamed schema

There is a NestedFieldsProjectableTableSource in Flink, but so far, no connector has implemented it. There may be unknown risks, and I tend not to implement it first.

Another case: Project with a new renamed schema

You mean: Create a table, insert some data, rename some fields, insert some data. Then read table using Flink?

Not the cases you said, I read the cases in TestReadProject, and was thought that we may also need a ut to address this one: write records into a table and then read them from table by another schema with same fields id but different field names. But we flink are prejection with a List<String> (field names), so we don't have such case. We could ignore the case for unit test.

Field reordering tests are at the file format level. Each file format has to be able to project columns in the requested order. So any reordered schema should work as long as it is passed down correctly, which is what the new convert method does.

openinx · 2020-08-18T08:36:23Z

flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java

+
+    Map<String, Integer> indexByName = TypeUtil.indexByName(schema.asStruct());
+    Set<Integer> projectedIds = projectedFields.stream().map(indexByName::get).collect(Collectors.toSet());
+    return TypeUtil.select(schema, projectedIds);


Continue with the question from here. If we could produce a ordered & projected schema in this method (Saying if this method is pruneWithReordering), then seems we don't have to convert the read RowData to the correct order here ?

I'd prefer to use the correct projected schema to read the target RowData if possible, rather than reading RowData in a disordered schema and then order them in an iterator transformation. Because this is in the critical read path and an extra RowData transformation will cost more resources , also make the codes hard to follow.

I'd prefer to use the correct projected schema to read the target RowData if possible

This is what I want too, but I'm afraid the current format readers do not have this capability. You can take a look to AvroSchemaWithTypeVisitor, the readers order is according to file schema instead of Flink projected/expected schema.

Because this is in the critical read path and an extra RowData transformation will cost more resources

The performance is OK, because we just use a lazy projection in ProjectionRowData, Unnecessary projections are omitted.

cc @rdblue , Mind to take a look this ? Thanks.

You should be able to use the expected/projection schema. All readers should reorder columns to produce the requested column order.

AvroSchemaWithTypeVisitor is used to traverse the file schema to create the reader structure, but that's because fields in Avro must be read in the file's order. But when that reader adds data columns to records, the values are put in the correct order because the ResolvingDecoder returns the correct position in the projection schema.

Thanks @openinx and @rdblue ! I'm very happy to be able to solve my confusion. I will do it.

It doesn't matter if we change the order of the schema, as long as the ID doesn't change.

rdblue · 2020-08-20T20:14:40Z

@JingsongLi, now that #1332 has been merged, could you update this one? It looks like this one is next to review.

JingsongLi · 2020-08-21T03:59:51Z

Hi @openinx and @rdblue , rebased and added case and fixed bug.
I think this is ready to review.

openinx · 2020-08-21T08:21:32Z

Fine, I will take a look.

openinx · 2020-08-21T09:43:28Z

flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java

+   * @return a Schema corresponding to the Flink projection
+   */
+  public static Schema projectWithReordering(Schema schema, List<String> projectedFields) {
+    if (projectedFields == null) {


nit : projectedFields == null || projectFields.length ==0

projectFields.length ==0 means project empty column. But projectedFields == null means project all columns.

openinx · 2020-08-21T09:49:30Z

flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java

+
+  /**
+   * Project columns from a {@link Schema} using a projected fields.
+   *


we may need to add a comment to indicate that: we don't support complex data type projection for flink now.

Add comments: Don't support nested fields projection for Flink now.

Why aren't nested fields supported?

Flink SQL did not have good support for nested projection before. I'll verify it and try to implement nested fields projection.

openinx · 2020-08-21T09:51:48Z

flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java

+   * @param projectedFields projected fields from Flink
+   * @return a Schema corresponding to the Flink projection
+   */
+  public static Schema projectWithReordering(Schema schema, List<String> projectedFields) {


We've already have a Schema#select, will it fit for your requirement ?

/** * Creates a projection schema for a subset of columns, selected by name. * <p> * Names that identify nested fields will select part or all of the field's top-level column. * * @param names a List of String names for selected columns * @return a projection schema from this schema, by name */ public Schema select(Collection<String> names) { return internalSelect(names, true); }

You can take a look to Schema.internalSelect, it is a Set<Integer> selected, so actually the interface is: "selectWithoutReordering"

OK, that sounds reasonable.

We do this in Spark here: https://github.com/apache/iceberg/blob/master/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java#L161-L166

You might try a similar approach since we already have Flink to Iceberg conversion. It should just be a matter of reassigning the IDs. You may also need the type fixes, I don't recall if Iceberg to Flink to Iceberg conversion is lossy or not.

You may also need the type fixes

Yes, we can have.

After thinking about it with nested projection, we can pass a TableSchema requestedSchema parameter, which contains the required projection (Including nested) and the type to be fixed.

Only UUID need be fixed...

Create #1382 for fixup.

openinx · 2020-08-21T09:54:51Z

flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

+ * under the License.
+ */
+
+package org.apache.iceberg.flink.source;


Here, you put all reader related classes inside the source package, will we also need to put those writer related classes into sink package ? I don't have strong feeling to do that, keeping consistence is OK for me.

will we also need to put those writer related classes into sink package ?

I think we can.

flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

openinx · 2020-08-21T10:06:40Z

flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

+      case BINARY: // byte[]
+        return ByteBuffers.toByteArray((ByteBuffer) value);
+      case TIME: // int instead of long
+        return (int) (DateTimeUtil.timeFromMicros((Long) value).toNanoOfDay() / 1000_000);


the long value is surely microseconds ? could we just return (int)((long)value/1_000) ?

openinx · 2020-08-21T10:11:53Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

+  public FlinkInputSplit[] createInputSplits(int minNumSplits) throws IOException {
+    // Called in Job manager, so it is OK to load table from catalog.
+    tableLoader.open(serializableConf.get());
+    try (TableLoader loader = tableLoader) {


Em, seems I've missed to close the TableLoader in IcebergFilesCommitter patch..

flink/src/main/java/org/apache/iceberg/flink/source/RowDataIterator.java

JingsongLi · 2020-08-24T02:25:45Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcReaders.java

    StructReader(List<OrcValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) {
      super(readers, struct, idToConstant);
-      this.numFields = readers.size();
+      this.numFields = struct.fields().size();


This fixes Flink Orc Reader (with partition) bug.

Nice catch, if the schema is a projected read schema, then the numFields will be mismatched.

rdblue · 2020-08-26T00:57:20Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcReader.java

    this.reader = OrcSchemaWithTypeVisitor.visit(iSchema, readSchema, new ReadBuilder(idToConstant));
  }

-  public static OrcRowReader<RowData> buildReader(Schema schema, TypeDescription readSchema) {


Was this not used anywhere?

Because there are two constructors, so we need to have two static helpers, I think we can use constructors directly.

It is just used by testing code.

rdblue · 2020-08-26T00:57:45Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcReaders.java

+
+      // The hive ORC writer may will adjust the scale of decimal data.
+      Preconditions.checkArgument(value.precision() <= precision,
+          "Cannot read value as decimal(%s,%s), too large: %s", precision, scale, value);


Does Flink require this?

I think it is better to add this check to avoid potential precision mismatched bugs.

rdblue · 2020-08-26T01:02:11Z

flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

+        return ByteBuffers.toByteArray((ByteBuffer) value);
+      case BINARY: // byte[]
+        return ByteBuffers.toByteArray((ByteBuffer) value);
+      case TIME: // int instead of long


Nit: int in millis?

rdblue · 2020-08-26T01:03:03Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

+ */
+public class FlinkInputFormat extends RichInputFormat<RowData, FlinkInputSplit> {
+
+  private static final long serialVersionUID = 1L;


Is this needed?

For a stream computing job, I think it's better to have it. If there is a Job running in the cluster, in the future, if user update Iceberg-Flink version, this version modify something that does not affect compatibility, but resulting in a change to serialVersionUID, the user's job will be incompatible after the cluster upgrade. In fact, this situation is compatible.

Is Flink using Java serialization across versions? That seems like a big risk to me. I'd prefer to only use Java serialization between processes running the exact same version of Iceberg. If we need to serialize across versions (like for checkpoint data) then I think we should worry about compatibility a lot more.

openinx · 2020-08-26T02:48:00Z

flink/src/main/java/org/apache/iceberg/flink/source/ScanOptions.java

+      return this;
+    }
+
+    public ScanOptions build() {


Q: do you think whether there's need to abstract the common options builder sharing between flink and spark (maybe also hive/pig) to validate and build those properties into a ScanOptions ? If sure, we may finish that in a new separate pr.

I don't know, there are some difference between Flink/Spark and Hive, maybe we can try to do something in Spark.

openinx · 2020-08-26T02:59:55Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java

+
+  @Test
+  public void testIdentityPartitionProjections() throws Exception {
+    Schema logSchema = new Schema(


Q: Do we need more cases to address the other partitioned data type in https://github.com/apache/iceberg/pull/1346/files#diff-84728688cba8556f9ff91f32d3873efcR112 ?

There is testPartitionTypes for this.

JingsongLi · 2020-08-26T07:01:48Z

Thanks @rdblue and @openinx for the review, I have added project(TableSchema schema) and select(List<String> fields) to format builder, project can support nested projection.

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

openinx · 2020-08-26T09:02:37Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

+        TableSchema.Builder builder = TableSchema.builder();
+        for (String field : selectedFields) {
+          TableColumn column = tableSchema.getTableColumn(field).orElseThrow(() -> new IllegalArgumentException(
+              "The fields are illegal in projectedFields: " + selectedFields));


I think we need to point out which column is missing in the error message .

openinx · 2020-08-26T09:49:39Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

+    public Builder icebergSchema(Schema newSchema) {
+      this.icebergSchema = newSchema;
+      return this;
+    }
+
+    public Builder io(FileIO newIO) {
+      this.io = newIO;
+      return this;
+    }
+
+    public Builder encryption(EncryptionManager newEncryption) {
+      this.encryption = newEncryption;
+      return this;
+    }


Do we really need those three methods ? I saw that we would loadTable and override all of the three if anyone is null, that says setting one or two of them won't work in this builder.

For me , the table(Table newTable) and tableLoader(TableLoader newLoader) is enough.

OK I can remove them.

rdblue · 2020-09-14T01:11:45Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+        encryption = table.encryption();
+      }
+
+      if (projectedSchema != null && selectedFields != null) {


Is this needed? If this delegated to when the scan is built, then the scan would do the check and users would get consistent exception messages.

See below comments.

rdblue · 2020-09-14T01:12:43Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+                  field, tableSchema)));
+          builder.field(column.getName(), column.getType());
+        }
+        projectedTableSchema = builder.build();


Why project the Flink schema manually rather than using icebergSchema.select(selectedFIelds) and converting the result?

I think we remove this select, because what Flink SQL want, is an order changed select, instead of using original iceberg table order.
We should provide a unified select, so I think we can provide in Flink side now, and we can use project.

rdblue · 2020-09-14T01:42:44Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+      return this;
+    }
+
+    public Builder project(TableSchema schema) {


Will SQL use this or select?

SQL use order changed select, because SQL not supports nested fields push down now.
But I think should support nested push down in future, then, should use project.

But if we provide an original schema ordered select(Set<String>) like TableScan.select, I think SQL can not use this one.

rdblue · 2020-09-14T01:45:06Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+      return env;
+    }
+
+    RowDataTypeInfo getRowTypeInfo() {


Since these are intended to be called from child classes, should they be protected?

rdblue · 2020-09-14T01:47:13Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+    public DataStream<RowData> build() {
+      Preconditions.checkNotNull(getEnv(), "StreamExecutionEnvironment should not be null");
+      FlinkInputFormat format = buildFormat();
+      return getEnv().createInput(format, getRowTypeInfo());


How is this different than an unbounded builder? I don't see anything that passes whether the stream should be bounded or unbounded. It seems like this should pass that information so that the input adapter can plan the current table scan, rather than checking for new data later.

Not having endSnapshot in ScanOptions seems to imply that the scan is unbounded.

I was thinking about adding a new config option like bounded is true or not.
After reading your comments, I think it's very good. Because streaming jobs have a high probability of setting the starting snapshot ID or timestamp, the strategy can be:

Without startSnapshotId: Bounded

With startSnapshotId and with endSnapshotId: Bounded

With startSnapshotId (-1 means unbounded preceding) and Without endSnapshotId: Unbounded

In this way, we can have a unify builder.

Sounds like a good way to configure this to me, except that we will want to make sure the default is reasonable. For Flink, should the default be unbounded or bounded?

I prefer that default is bounded.

Bounded mode is more common in attempt and startup of users.

In unbounded mode, users often define startSnapshotId . But in bounded mode, endSnapshotID is rare. If default is unbounded, it is hard to define bounded mode.

rdblue · 2020-09-14T01:51:34Z

flink/src/main/java/org/apache/iceberg/flink/source/ScanOptions.java

+    return caseSensitive;
+  }
+
+  public Long getSnapshotId() {


I'm not sure if this class should conform to the typical style used by Flink or Iceberg, but in Iceberg, we omit get from getter names because it doesn't add any helpful context and is awkward in non-Java languages where getter methods are named for fields.

rdblue · 2020-09-14T17:34:27Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkInputFormat.java

+      Table table, List<String> projectFields, CatalogLoader loader, Long snapshotId, Long startSnapshotId,
+      Long endSnapshotId, Long asOfTimestamp, List<Expression> filters, String sqlFilter) throws IOException {
+    ScanOptions options = ScanOptions.builder().snapshotId(snapshotId).startSnapshotId(startSnapshotId)
+        .endSnapshotId(endSnapshotId).asOfTimestamp(asOfTimestamp).build();


I think the tests would be much more readable if the ScanOptions builder were used directly. For example, this is hard to understand:

assertRecords(executeWithOptions(table, null, null, null, snapshotId1, null, null, null, null), expected1, SCHEMA);

But you could rewrite that like this:

ScanOptions options = ScanOptions.builder().startSnapshotId(snapshotId1).build(); assertRecords(executeWithOptions(table, options), expected1, SCHEMA);

In addition, passing null into this leaks the default state within the builder into the tests: test authors need to know that passing null for CatalogLoader is supported. I think it is better to let the test authors use the builder pattern.

Agree with you.
I was thinking about how to make it easy for SQL tests to reuse it. SQL testing can also rebuild SQL strings from ScanOptions.

rdblue · 2020-09-25T00:53:29Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java

+    long timestampMillis = table.currentSnapshot().timestampMillis();
+
+    // produce another timestamp
+    Thread.sleep(10);


We usually introduce a spin to avoid sleeping for long durations in lots of tests, like this: https://github.com/apache/iceberg/blob/master/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java#L78

rdblue · 2020-09-25T01:02:32Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java

+   * The Flink SQL output: [f2, data]
+   * The FlinkInputFormat output: [nested[f2], data].
+   */
+  protected abstract void assertNestedProjection(Table table, List<Record> records) throws IOException;


I find it really strange that this is delegated to a subclass, given that it builds a very specific nested projection.

Why not make this use a method like execute(Table, List<String>), but pass in the projection instead of a list of fields?

Then you could keep all of the schema details in the test method here, rather than delegating this assertion. I think it would be cleaner.

I think we can move testNestedProjection to TestFlinkInputFormat, because only InputFormat supports nested push-down, SQL can not.

rdblue · 2020-09-25T01:03:35Z

flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java

+   * But the FlinkInputFormat can't.
+   */
+  protected abstract void assertResiduals(Schema schema, List<Row> results, List<Record> writeRecords,
+                                          List<Record> filteredRecords) throws IOException;


Why is this implemented by the subclass? Couldn't this just call assertRecords directly?

This is also because of the difference between InputFormat and SQL(It can actually filter out the data). I think I should put it into a subclass of SQL.

rdblue · 2020-09-25T01:10:06Z

I'm going ahead and merging this because I think the remaining things are minor, but we will probably want to fix them.

First, breaking the tests across two classes could be cleaner. In the future, I'd recommend writing tests and not breaking across a parent/child class until you want to add the other set of tests. That way it is easier to see what is needed and will be clean.

Second, I don't understand the benefit of having a separate ScanOptions class, besides that it can pull options out of a map. It seems to me that it would be simpler to just have the FlinkSource.Builder class and move all of the scan options to that class.

JingsongLi · 2020-09-25T02:20:04Z

Thanks @rdblue for the review, I'll address your test comments in next PR (Integrate to SQL).
The reason for a separate ScanOptions is: For SQL layer, ScanOptions contains options in CREATE TABLE ... WITH (options), while other parameters of FlinkSource.Builder are not. This means that the parameters in ScanOptions must be in the form of string.
If you don't think it is necessary, we can also merge it into the builder.

rdblue · 2020-09-25T17:30:02Z

Let's take a look at ScanOptions in the next PR then. I would prefer to keep user-facing APIs simple, rather than leaking a SQL concern (options come from WITH) to users (need to use two builders). Since SQL will most likely use the fromProperties method, it may make sense to use a single builder, add withProperties, and pass properties from SQL as a map.

Jiayi-Liao · 2020-10-19T07:49:46Z

flink/src/main/java/org/apache/iceberg/flink/source/FlinkSource.java

+   *
+   * @return {@link Builder} to connect the iceberg table.
+   */
+  public static Builder forRowData() {


Hi @JingsongLi , I'm testing Iceberg recently. Since the StreamExecutionEnvironment is a must-have parameter for FlinkSource, would it better to put it in the builder's constructor instead of FlinkSource.forRowData().env(xx)?

just a minor improvement on user experience.

I think it is better to keep builder pattern. And for now, we can create a FlinkInputFormat without env too.

I see... The builder is for both InputFormat and DataStreamSource.

JingsongLi mentioned this pull request Aug 17, 2020

Flink: Implement Flink InputFormat and integrate it to FlinkCatalog #1293

Closed

JingsongLi force-pushed the format branch from f5f7d75 to decf8f1 Compare August 17, 2020 08:43

openinx reviewed Aug 18, 2020

View reviewed changes

JingsongLi force-pushed the format branch from decf8f1 to d327aa8 Compare August 20, 2020 04:40

probot-autolabeler bot added the flink label Aug 20, 2020

JingsongLi force-pushed the format branch from 2f53fd9 to 0b27405 Compare August 20, 2020 05:25

JingsongLi force-pushed the format branch from 0b27405 to 4259635 Compare August 21, 2020 02:43

openinx reviewed Aug 21, 2020

View reviewed changes

flink/src/main/java/org/apache/iceberg/flink/source/RowDataIterator.java Show resolved Hide resolved

JingsongLi commented Aug 24, 2020

View reviewed changes

openinx approved these changes Aug 24, 2020

View reviewed changes

rdblue reviewed Aug 26, 2020

View reviewed changes

openinx reviewed Aug 26, 2020

View reviewed changes

probot-autolabeler bot added API spark labels Aug 26, 2020

JingsongLi mentioned this pull request Aug 26, 2020

Flink: Support Flink streaming reading #1383

Closed

openinx reviewed Aug 26, 2020

View reviewed changes

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java Outdated Show resolved Hide resolved

openinx reviewed Aug 26, 2020

View reviewed changes

rdblue reviewed Sep 14, 2020

View reviewed changes

chenjunjiedada mentioned this pull request Sep 14, 2020

Flink: consider add serialVersionUID for serializable classes used by operator #1438

Closed

rdblue reviewed Sep 14, 2020

View reviewed changes

JingsongLi force-pushed the format branch from 43309b1 to 69e753a Compare September 15, 2020 06:44

JingsongLi added 7 commits September 24, 2020 10:19

Flink: Introduce Flink InputFormat

e0fce7f

Address comment

b6e9ecd

checkstyles

b13ed9b

Introduce FlinkSource

8d080ad

Checkstyle

3c727fa

tmp

6c5f830

Address comments

e27e421

JingsongLi force-pushed the format branch from 69e753a to e27e421 Compare September 24, 2020 02:19

Set tasks to null in DataIterator

45af8c7

rdblue reviewed Sep 25, 2020

View reviewed changes

rdblue merged commit 1a797cd into apache:master Sep 25, 2020

JingsongLi mentioned this pull request Sep 25, 2020

Flink: Integrate Flink reader to SQL #1509

Merged

Jiayi-Liao reviewed Oct 19, 2020

View reviewed changes

JingsongLi deleted the format branch November 5, 2020 09:42

rdblue added this to the Java 0.10.0 Release milestone Nov 16, 2020

openinx mentioned this pull request Jan 7, 2021

Flink: Support streaming reader. #1793

Merged

Flink: Introduce Flink InputFormat #1346

Flink: Introduce Flink InputFormat #1346

Uh oh!

Conversation

JingsongLi commented Aug 17, 2020 • edited by rdblue Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue Aug 20, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Aug 20, 2020

Uh oh!

JingsongLi commented Aug 21, 2020

Uh oh!

openinx commented Aug 21, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

JingsongLi commented Aug 17, 2020 •

edited by rdblue

Loading

rdblue Aug 20, 2020 •

edited

Loading

JingsongLi Aug 24, 2020 •

edited

Loading

JingsongLi Aug 26, 2020 •

edited

Loading